package com.ai;

import com.ai.aif.msgframe.common.io.MsgSerialize;
import com.ai.aif.msgframe.common.message.MsgFMessage;
import com.ai.aif.msgframe.common.message.MsgFTextMessage;
import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

import java.util.Properties;

import static com.ai.aif.msgframe.common.util.MessageCovertUtil.createMsgSerialize;

/**
 * Created by yinpeng on 16/11/25.
 */
public class MKProducerClient {

    public void  send(String subject, MsgFTextMessage message){

        Properties props = new Properties();
        //此处配置的是kafka的端口
        props.put("metadata.broker.list","10.1.236.190:9092,10.1.236.191:9092,10.1.236.192:9092");
        props.put("serializer.class", "kafka.serializer.DefaultEncoder");
        props.put("partitioner.class", "com.ai.Perfermanceartitioner");
        props.put("compression.type", "gzip");
        props.put("key.serializer.class", "kafka.serializer.StringEncoder");
        props.put("group.id", "id1");
        props.put("request.required.acks", "1");
        Producer producer = new Producer(new ProducerConfig(props));
        // 构建消息体
            producer.send(transKafkaMessage(subject,message));
        producer.close();
    }



    public static KeyedMessage transKafkaMessage(String topic, MsgFMessage message)
{
		 MsgSerialize serialize = createMsgSerialize();
		KeyedMessage record = new KeyedMessage(topic, message.getMsgId(),
                serialize.serialize(message));
		return record;
		}
}
